package common

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
)

const (
	EMPTY     = "EMPTY"     //无数据
	HEARTBEAT = "HEARTBEAT" //心跳
)

type MsgType int32

const (
	POST_WAIT_RESPONSE MsgType = iota // 请求，并等待服务器响应（客户端推送消息后需要等待并接收服务端的处理结果）
	POST_NO_RESPONSE                  // 请求，不需要服务器响应（客户端推送消息后不需要等待服务端的处理结果，防止网络I/O阻塞）
	RESPONSE                          // 服务器响应
	QUERY                             // 查询消息队列
	QUERY_POP                         // 查询消息队列，并删除消息
	QUERY_STREAM                      // 查询消息队列，返回消息流，每个消息只会被一个消费者消费
	QUERY_STREAM_GROUP                // 查询消息队列，返回消息流，每个消息可以被多个消费者消费
)

// NeedReplyToClient 是否需要向客户端响应
func (mt MsgType) NeedReplyToClient() bool {
	return mt != POST_NO_RESPONSE
}

// IsPost 是否是post类型的消息
func (mt MsgType) IsPost() bool {
	if mt == POST_WAIT_RESPONSE || mt == POST_NO_RESPONSE {
		return true
	}
	return false
}

// IsQueryStream 是否是query stream类型的消息
func (mt MsgType) IsQueryStream() bool {
	if mt == QUERY_STREAM || mt == QUERY_STREAM_GROUP {
		return true
	}
	return false
}

// IsQuery 是否是query类型的消息
func (mt MsgType) IsQuery() bool {
	if mt == QUERY || mt == QUERY_POP || mt.IsQueryStream() {
		return true
	}
	return false
}

var msgTypeStr = map[MsgType]string{
	POST_WAIT_RESPONSE: "POST_WAIT_RESPONSE",
	POST_NO_RESPONSE:   "POST_NO_RESPONSE",
	RESPONSE:           "RESPONSE",
	QUERY:              "QUERY",
	QUERY_POP:          "QUERY_POP",
	QUERY_STREAM:       "QUERY_STREAM",
	QUERY_STREAM_GROUP: "QUERY_STREAM_GROUP",
}

func (msgType MsgType) String() string {
	return msgTypeStr[msgType]
}

const (
	minValidMsgLength = 28
	msgHeaderLen      = 20

	Version = 1
)

func SuccessMsg(topic string) *Msg {
	ret := ResponseMsg(topic, []byte("SUCCESS"))
	return ret
}

func FailMsg(topic string, errorInfo string) *Msg {
	ret := ResponseMsg(topic, []byte("FAIL:"+errorInfo))
	return ret
}

func ResponseMsg(topic string, payload []byte) *Msg {
	ret := NewMsg(MsgHeader{
		Version: Version,
		Type:    RESPONSE,
	}, MsgBody{
		Topic:   []byte(topic),
		Payload: payload,
	})
	return ret
}

type Msg struct {
	Header MsgHeader
	Body   MsgBody
}

type MsgHeader struct {
	Version        uint32
	Type           MsgType
	BodyLen        uint32 // 消息体长度
	ExpireAfterNow uint32 // 相对有效期，单位ms，真实的有效截至时间=消息接收方的now() + ExpireAfterNow, 0表示永久有效
	ClientId       uint32 // 客户端标识
}

func NewMsg(Header MsgHeader,
	Body MsgBody) *Msg {
	m := &Msg{
		Header: Header,
		Body:   Body,
	}
	m.InitLength()
	return m
}

func (msg *Msg) InitLength() {
	msg.Body.TopicLen = uint32(len(msg.Body.Topic))
	msg.Body.PayloadLen = uint32(len(msg.Body.Payload))
	msg.Header.BodyLen = msg.Body.TopicLen + msg.Body.PayloadLen + 8
}

type MsgBody struct {
	TopicLen   uint32
	PayloadLen uint32 // Payload长度
	Topic      []byte

	Payload []byte // 消息内容
}

func ReadMsg(reader io.Reader) (*Msg, error) {
	buf := make([]byte, msgHeaderLen)
	_, err := io.ReadFull(reader, buf)
	if err != nil {
		return nil, err
	}
	header, err := decodeHeader(buf)
	if err != nil {
		return nil, err
	}

	buf = make([]byte, header.BodyLen)
	_, err = io.ReadFull(reader, buf)
	if err != nil {
		return nil, err
	}

	body, err := decodeBody(buf)
	if err != nil {
		return nil, err
	}

	var msg Msg
	msg.Header = header
	msg.Body = body
	return &msg, nil
}

func decodeHeader(b []byte) (MsgHeader, error) {
	var header MsgHeader
	header.Version = binary.BigEndian.Uint32(b[:4])
	header.Type = MsgType(binary.BigEndian.Uint32(b[4:8]))
	header.BodyLen = binary.BigEndian.Uint32(b[8:12])
	header.ExpireAfterNow = binary.BigEndian.Uint32(b[12:16])
	header.ClientId = binary.BigEndian.Uint32(b[16:20])
	return header, nil
}

func decodeBody(b []byte) (MsgBody, error) {
	var body MsgBody

	body.TopicLen = binary.BigEndian.Uint32(b[0:4])
	body.PayloadLen = binary.BigEndian.Uint32(b[4:8])

	topicEnd := 8 + body.TopicLen
	if body.TopicLen > 0 {
		body.Topic = make([]byte, body.TopicLen)
		copy(body.Topic, b[8:topicEnd])
	}

	if len(b[topicEnd:]) != int(body.PayloadLen) {
		return body, fmt.Errorf("invalid message payload length, should be: %d, actual: %d", body.PayloadLen, len(b[topicEnd:]))
	}

	if body.PayloadLen > 0 {
		body.Payload = make([]byte, body.PayloadLen)
		copy(body.Payload, b[topicEnd:])
	}

	return body, nil
}

func decodeMessage(b []byte) (*Msg, error) {
	var msg Msg

	if len(b) < minValidMsgLength {
		return nil, fmt.Errorf("invalid message byte size (%d)", len(b))
	}

	header, err := decodeHeader(b[:msgHeaderLen])
	if err != nil {
		return nil, err
	}

	body, err := decodeBody(b[msgHeaderLen:])
	if err != nil {
		return nil, err
	}

	msg.Header = header
	msg.Body = body

	return &msg, nil
}

func (m *Msg) ByteSize() uint32 {
	return m.Header.BodyLen + msgHeaderLen
}

// WriteTo 将消息写入writer
func (m *Msg) WriteTo(w io.Writer) (int64, error) {
	var buf [msgHeaderLen]byte
	var total int64

	binary.BigEndian.PutUint32(buf[:4], m.Header.Version)
	binary.BigEndian.PutUint32(buf[4:8], uint32(m.Header.Type))
	binary.BigEndian.PutUint32(buf[8:12], m.Header.BodyLen)
	binary.BigEndian.PutUint32(buf[12:16], m.Header.ExpireAfterNow)
	binary.BigEndian.PutUint32(buf[16:20], m.Header.ClientId)

	n, err := w.Write(buf[:])
	total += int64(n)
	if err != nil {
		return total, err
	}

	var buf2 [8]byte
	binary.BigEndian.PutUint32(buf2[0:4], m.Body.TopicLen)
	binary.BigEndian.PutUint32(buf2[4:8], m.Body.PayloadLen)
	n, err = w.Write(buf2[:])
	total += int64(n)
	if err != nil {
		return total, err
	}

	n, err = w.Write(m.Body.Topic[:])
	total += int64(n)
	if err != nil {
		return total, err
	}

	n, err = w.Write(m.Body.Payload[:])
	total += int64(n)
	if err != nil {
		return total, err
	}

	return total, nil
}

func (msg *Msg) Bytes() []byte {
	bf := bytes.NewBuffer(nil)
	msg.WriteTo(bf)
	return bf.Bytes()
}

func (msg *Msg) IsHeartbeat() bool {
	return string(msg.Body.Payload) == HEARTBEAT
}

func (msg *Msg) String() string {
	return fmt.Sprintf("[Version:%d, Type:%s, ClientId:%d, Topic:%s, Payload:%s]",
		msg.Header.Version, msg.Header.Type, msg.Header.ClientId, string(msg.Body.Topic), string(msg.Body.Payload))
}

func (msg *Msg) Copy() *Msg {
	cp := &Msg{
		Header: MsgHeader{
			Version:        msg.Header.Version,
			Type:           msg.Header.Type,
			BodyLen:        msg.Header.BodyLen,
			ExpireAfterNow: msg.Header.ExpireAfterNow,
			ClientId:       msg.Header.ClientId,
		},
		Body: MsgBody{
			TopicLen:   msg.Body.TopicLen,
			PayloadLen: msg.Body.PayloadLen,
			Topic:      make([]byte, msg.Body.TopicLen),
			Payload:    make([]byte, msg.Body.PayloadLen),
		},
	}
	copy(cp.Body.Topic, msg.Body.Topic)
	copy(cp.Body.Payload, msg.Body.Payload)
	return cp
}
